package com.lagou.smartlog.writer

import java.sql.{Connection, PreparedStatement}
import java.util.Properties

import com.alibaba.druid.pool.{DruidDataSource, DruidDataSourceFactory}
import javax.sql.DataSource
import org.apache.spark.sql.ForeachWriter

object MysqlWriter{

    val pro: Properties = new Properties()
    pro.load(MysqlWriter.getClass.getClassLoader.getResourceAsStream("jdbc.properties"))

    val dataSource = DruidDataSourceFactory.createDataSource(pro)
    private val threadLocal: ThreadLocal[Connection] = new ThreadLocal

    def getConnection() = {

        var connection: Connection = threadLocal.get()
        if (connection == null){
            connection= dataSource.getConnection()
            threadLocal.set(connection)
        }

        connection
    }
}

class MysqlWriter extends ForeachWriter[BusInfo]{
    var conn: Connection = _

    //开启连接
    override def open(partitionId: Long, epochId: Long): Boolean = {
        conn = MysqlWriter.getConnection()
        true
    }

    //把数据写入mysql
    override def process(value: BusInfo): Unit = {
        //把数据写入mysql
        val lglat: String = value.lglat
        val deployNum = value.deployNum

        val sql:String="insert into position values(?,?)"

        val pst: PreparedStatement = conn.prepareStatement(sql)
        pst.setString(1,deployNum)
        pst.setString(2,lglat)

        pst.executeUpdate()
    }

    //释放连接
    override def close(errorOrNull: Throwable): Unit = {
        conn.close()
    }
}
